package com.cat.net.network.kafka;

import com.cat.net.exception.RepeatProtoException;
import com.cat.net.network.annotation.Kmd;
import com.cat.net.network.base.*;
import com.cat.net.network.controller.IRpcController;
import com.cat.net.util.SerializationUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * kafka消息调度器<br>
 * 调度器的仅用于解析消息, 处理消息. 不需要关注处理顺序, kafka的消息分区, 取出消息已经是有序的了.
 * 也就意味着, 处理时不需要关心来自不同分区的消息间的顺序, 如果要跨分区保持全局顺序, 那么需要重新考虑架构或者使用单分区来保证顺序.
 */
public class KafkaDispatcher {

    public final Logger log = LoggerFactory.getLogger(this.getClass());
    
    /**
     * key: 消息号
     * value: 消息封装
     */
    protected Map<Integer, RemoteServerCaller> mapper = new HashMap<>();

    public RemoteCaller getRemoteCaller(int cmd) {
        return mapper.get(cmd);
    }

    /**
     * 初始化
     * @param controllers 消息处理接口列表
     * @throws Exception 异常
     */
    public void initialize(List<IRpcController> controllers) throws Exception {
        long startTime = System.currentTimeMillis();
        for (IRpcController controller : controllers) {
            Method[] methods = controller.getClass().getDeclaredMethods();
            for (Method method : methods) {
                Kmd cmd = method.getAnnotation(Kmd.class);
                if (cmd == null) {
                    continue;
                }
                //检查重复协议号
                if (mapper.containsKey(cmd.value())) {
                    throw new RepeatProtoException("发现重复协议号:"+cmd.value());
                }
                mapper.put(cmd.value(), RemoteServerCaller.create(controller, false, method));
            }
        }
        log.info("The initialization message[{}] is complete and takes [{}] milliseconds.", mapper.size(),(System.currentTimeMillis() - startTime));
    }

    /**
     * 当收到消息
     * @param packet 消息体
     */
    public void onReceiveMessage(Packet packet) {
        int cmd = packet.cmd();
        if (cmd < 0) {
            log.info("Received inner message, cmd:[{}]", cmd);
            return;
        }
        RemoteServerCaller commander = mapper.get(cmd);
        if (commander == null) {
            log.info("收到未处理协议, cmd=[{}]", cmd);
            return;
        }
        this.invoke(commander, packet);
    }

    /**
     * 调度
     * @param caller 调用方法
     * @param packet 包
     */
    public void invoke(RemoteServerCaller caller, Packet packet)  {
        int cmd = packet.cmd();
        long beginTime = System.currentTimeMillis();

        Class<?> clazz = caller.getParamType();
        AbstractProtocol params = (AbstractProtocol) SerializationUtil.deserialize(packet.data(), clazz);
        //处理消息执行
        caller.getInvoker().invoke(params);
        long used = System.currentTimeMillis() - beginTime;
        // 协议处理超过1秒
        if (used > 1000) {
            log.info("协议[{}]处理慢!!!耗时{}ms", cmd, used);
        }
    }

}
